 1.Kafka高级特性解析之原理剖析
   
   由上图可以看出：KafkaProducer有两个基本线程：
   1).主线程：负责消息创建，拦截器，序列化器，分区器等操作，并将消息追加到消息收集器
RecoderAccumulator中；
   消息收集器RecoderAccumulator为每个分区都维护了⼀个 Deque<ProducerBatch> 类型的双端队列。
   ProducerBatch 可以理解为是 ProducerRecord 的集合，批量发送有利于提升吞吐量，降低网络影响；
   由于生产者客户端使用java.io.ByteBuffer 在发送消息之前进⾏消息保存，并维护了一个
BufferPool 实现 ByteBuffer 的复用；该缓存池只针对特定大小（ batch.size 指定）的 ByteBuffer
进行管理，对于消息过⼤的缓存，不能做到重复利用。
   每次追加一条ProducerRecord消息，会寻找/新建对应的双端队列，从其尾部获取一个
ProducerBatch，判断当前消息的大小是否可以写入该批次中。若可以写入则写入；若不可以写
入，则新建一个ProducerBatch，判断该消息⼤⼩是否超过客户端参数配置 batch.size 的值，不超
过，则以 batch.size建立新的ProducerBatch，这样方便进行缓存重复利用；若超过，则以计算的
消息大小建立对应的 ProducerBatch ，缺点就是该内存不能被复用了。
   2).Sender线程：
   该线程从消息收集器获取缓存的消息，将其处理为 <Node, List<ProducerBatch> 的形式， Node
表示集群的broker节点。
   进一步将<Node, List<ProducerBatch>转化为<Node, Request>形式，此时才可以向服务端发送
数据。
   在发送之前，Sender线程将消息以 Map<NodeId, Deque<Request>> 的形式保存到
InFlightRequests 中进行缓存，可以通过其获取 leastLoadedNode ,即当前Node中负载压力最小
的一个，以实现消息的尽快发出。
 
 2.Kafka高级特性解析之生产者参数配置补充
   
   1).参数设置方式：
   Map<String, Object> configs = new HashMap<String, Object>();
   configs.put("bootstrap.servers", "linux121:9092");
   //configs.put("key.serializer", IntegerSerializer.class);
   //configs.put("value.serializer", StringSerializer.class);
   configs.put("key.serializer", "org.apache.kafka.common.serialization.IntegerSerializer");
   configs.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
   configs.put("acks", "all");
   configs.put("compression.type", "gzip");
   configs.put("retries", 3);
   KafkaProducer<Integer, String> producer = new KafkaProducer<Integer, String>(configs);
   
   // 保证等待确认的消息只有设置的这几个，如果设置为1，则只有一个请求在等待响应
   // 此时可以保证发送消息即使在重试的情况下也是有序的
   configs.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 1);
   //configs.put("max.in.flight.request.per.connection", 1);
   2).补充参数：
   参数名称                   描述
   retry.backoff.ms           在向一个指定的主题分区重发消息的时候，重试之间的等待时间。
                              比如3次重试，每次重试之后等待该时间长度，再接着重试。在一些失败的场景，避免了密集循环的重新发送请求。
							  long型值，默认100。可选值：[0,...]
   retries                    retries重试次数
                              当消息发送出现错误的时候，系统会重发消息。
                              跟客户端收到错误时重发一样。
                              如果设置了重试，还想保证消息的有序性，需要设置MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION=1
                              否则在重试此失败消息的时候，其他的消息可能发送成功了
   request.timeout.ms         客户端等待请求响应的最⼤时⻓。如果服务端响应超时，则会重发请求，除⾮达到重试次数。该设置应该比
                              replica.lag.time.max.ms (a broker configuration)要大，以免在服务器延迟时间内重发消息。
							  int类型值，默认：30000，可选值：[0,...]
   interceptor.classes        在生产者接收到该消息，向Kafka集群传输之前，由序列化器处理之前，可以通过拦截器对消息进行处理。
                              要求拦截器类必须实现 org.apache.kafka.clients.producer.ProducerInterceptor 接⼝。
                              默认没有拦截器。
                              Map<String, Object> configs中通过List集合配置多个拦截器类名。
   acks                       当生产者发送消息之后，如何确认消息已经发送成功了。
                              支持的值：
							  acks=0：
                              如果设置为0，表示生产者不会等待broker对消息的确认，只要将消息放到缓冲区，就认为消息已经发送完成。
                              该情形不能保证broker是否真的收到了消息，retries配置也不会生效，因为客户端不需要知道消息是否发送成功。
                              发送的消息的返回的消息偏移量永远是-1。
							  acks=1
                              表示消息只需要写到主分区即可，然后就响应客户端，而不等待副本分区的确认。
                              在该情形下，如果主分区收到消息确认之后就宕机了，而副本分区还没来得及同步该消息，则该消息丢失。
							  acks=all
                              首领分区会等待所有的ISR副本分区确认记录。
                              该处理保证了只要有一个ISR副本分区存货，消息就不会丢失。
                              这是Kafka最强的可靠性保证，等效于 acks=-1 。
   batch.size                 当多个消息发送到同一个分区的时候，生产者尝试将多个记录作为一个批来处理。
                              批处理提高了客户端和服务器的处理效率。该配置项以字节为单位控制默认批的大小。
                              所有的批小于等于该值。
                              发送给broker的请求将包含多个批次，每个分区一个，并包含可发送的数据。 
                              如果该值设置的比较小，会限制吞吐量（设置为0会完全禁用批处理）。如果设置的很大，又有一点浪费内存，
                              因为Kafka会永远分配这么大的内存来参与到消息的批整合中。							  
   client.id                  生产者发送请求的时候传递给broker的id字符串。
                              用于在broker的请求⽇志中追踪什么应用发送了什么消息。
                              一般该id是跟业务有关的字符串。
   compression.type           生产者发送的所有数据的压缩方式。默认是none，也就是不压缩。
                              支持的值：none、gzip、snappy和lz4。
                              压缩是对于整个批来讲的，所以批处理的效率也会影响到压缩的比例。
   send.buffer.bytes          TCP发送数据的时候使⽤的缓冲区（SO_SNDBUF）大小。如果设置为0，则使用操作系统默认的。
   buffer.memory              生产者可以用来缓存等待发送到服务器的记录的总内存字节。如果记录的发送速度超过了将记录发送到服务器的速度，则生产者将阻
                              塞 max.block.ms 的时间，此后它将引发异常。此设置应⼤致对应于生产者将使用的总内存，但并非生产者使用的所有内存都用于
                              缓冲。一些额外的内存将⽤于压缩（如果启用了压缩）以及维护运行中的请求。long型数据。默认值：33554432，可选值：[0,...]
   connections.max.idle.ms    当连接空闲时间达到这个值，就关闭连接。long型数据，默认：540000
   linger.ms                  生产者在发送请求传输间隔会对需要发送的消息进⾏累积，然后作为一个批次发送。⼀般情况是消息的发送的速度比消息累积的速度
                              慢。有时客户端需要减少请求的次数，即使是在发送负载不大的情况下。该配置设置了一个延迟，生产者不会立即将消息发送到
                              broker，而是等待这么一段时间以累积消息，然后将这段时间之内的消息作为一个批次发送。该设置是批处理的另一个上限：一旦批
                              消息达到了 batch.size 指定的值，消息批会立即发送，如果积累的消息字节数达不到 batch.size 的值，可以设置该毫秒值，等
                              待这么长时间之后，也会发送消息批。该属性默认值是0（没有延迟）。如果设置 linger.ms=5 ，则在⼀个请求发送之前先等待
                              5 ms.long型值，默认：0，可选值：[0,...]
   max.block.ms               控制 KafkaProducer.send() 和 KafkaProducer.partitionsFor() 阻塞的时长。当缓存满了或元数据不可用的时候，这些
                              方法阻塞。在用户提供的序列化器和分区器的阻塞时间不计入。long型值，默认：60000，可选值：[0,...]
   max.request.size           单个请求的最⼤字节数。该设置会限制单个请求中消息批的消息个数，以免单个请求发送太多的数据。服务器有自己的限制批大小的
                              设置，与该配置可能不一样。int类型值，默认1048576，可选值：[0,...]
   partitioner.class          实现了接口 org.apache.kafka.clients.producer.Partitioner 的分区器实现类。默认值为：org.apache.kafka.
                              clients.producer.internals.DefaultPartitioner
   receive.buffer.bytes       TCP接收缓存（SO_RCVBUF），如果设置为-1,则使用操作系统默认的值。int类型值，默认32768，可选值：[-1,...]
   security.protocol          跟broker通信的协议：PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL.
                              string类型值，默认：PLAINTEXT
   max.in.flight.requests.    单个连接上未确认请求的最大数量。达到这个数量，客户端阻塞。如果该值大于1，且存在失败的请求，
    per.connection            在重试的时候消息顺序不能保证。
                              int类型值，默认5。可选值：[1,...]
   reconnect.backoff.max.ms   对于每个连续的连接失败，每台主机的退避将成倍增加，直至达到此最大值。在计算退避增量之后，
                              添加20％的随机抖动以避免连接风暴。
							  long型值，默认1000，可选值：[0,...]
   reconnect.backoff.ms       尝试重连指定主机的基础等待时间。避免了到该主机的密集重连。该退避时间应用于该客户端到broker的所有连接。
                              long型值，默认50。可选值：[0,...]
   